package keter.ssh.exec;

import java.io.File;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Lists;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;

import keter.framework.AbstractBaseSSHConfig;
import keter.framework.util.IOUtil;
import keter.ssh.core.SshUtil;
import keter.ssh.model.Host;

public class Executor {
	
	private static final Logger logger = LoggerFactory.getLogger(Executor.class);

	static List<SshUtil> sshList;
	
	static ListeningExecutorService pool;
	
	static List<ListenableFuture<String>> futures;
	
	public static void  init(List<Host> hosts,int thread) throws Exception {
 		pool = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(thread));
		sshList = Lists.newArrayList();
		for (Host host : hosts) {
			SshUtil ssh = SshUtil.init(host);
			sshList.add(ssh);
		}
	}
		
	public  static void execute(String cmd) throws InterruptedException{
		Validate.noNullElements(sshList,"ssh清单不能为空！"); 
		futures = Lists.newArrayList();
		for (SshUtil ssh : sshList) {
			ListenableFuture<String> future = pool.submit(new Callable<String>() {
			    @Override
			    public String call() throws Exception {
			    	return exec(cmd, ssh);
			    }
			});
			futures.add(future);
		}
		//顺序返回结果，空值或错误使用null替代
		ListenableFuture<List<String>> successfulQueries = Futures.successfulAsList(futures);
		Futures.addCallback(successfulQueries, new FutureCallback<Object>() {
			  @Override
			  public void onSuccess(Object result) {
			      System.out.println(result);
			  }

			  @Override
			  public void onFailure(Throwable t) {
			      //业务方法内部自行处理
			  }
	    });
//		logger.info("***poll to shutdown...***");
//		Futures
//		pool.awaitTermination(60, TimeUnit.SECONDS);
//		logger.info("***poll to shutdown...***");
		pool.shutdown();
	}
	
	private static String exec(String cmd, SshUtil ssh){
		try {
			return "\n--------  "+ssh.getHost().getAddr()+"  ------------\n"
		           +ssh.exec(cmd).result()
		           +"\n--------------------------------------";
		} catch (Exception e) {
			logger.error("主机:"+ssh.getHost().getAddr()+"命令执行失败!",e);
		}
		return "failure";
	}
	
	public static void download(String src,String dest){
		for (SshUtil ssh : sshList) {
			ListenableFuture<String> future = pool.submit(new Callable<String>() {
			    @Override
			    public String call() throws Exception {
			    	ssh.download(src, dest);
			    	return "success";
			    }
			});

			Futures.addCallback(future, new FutureCallback<String>() {
			  @Override
			  public void onSuccess(String result) {
				  logger.info("主机{}文件传输完成!",ssh.getHost().getAddr());
			  }

			  @Override
			  public void onFailure(Throwable t) {
			      logger.error("主机:"+ssh.getHost().getAddr()+"命令执行失败!",t);
			  }
	        });
		}
	}
	
	/**
	 * 将文件下载到各自主机目录内
	 * @param src
	 * @param dest
	 * @throws CmdException 
	 * @author gulixing@msn.com
	 * @date   2016年7月17日
	 */
	public static void downloadToEachFolder(String src,String dest){
		for (SshUtil ssh : sshList) {
			try {
				IOUtil.ensureExists(dest+"/"+ssh.getHost().getAddr());
				ssh.download(src, dest+"/"+ssh.getHost().getAddr());
				logger.info("主机{}文件传输完成，目标文件:{}",
					ssh.getHost().getAddr(),
					dest+File.separator+ssh.getHost().getAddr()+File.separator+StringUtils.substringAfterLast(src, "/"));
			} catch (Exception e) {
			      logger.error("主机:"+ssh.getHost().getAddr()+"命令执行失败!",e);
			}
		}
	}
	
//	@Deprecated
//	public  static void execute(String cmd) throws InterruptedException{
//	Validate.noNullElements(sshList,"ssh清单不能为空！"); 
//	for (SshUtil ssh : sshList) {
//		//多线程模式
//		if(AbstractBaseSSHConfig.prop.getInt("thread")>1){
//			conExec(cmd, ssh);
//		}
//		else{
//			//单线程模式
//			logger.info("\n------result of ["+ssh.getHost().getAddr()+"]:----------");
//		    System.out.println(exec(cmd, ssh)+"\n---------------------");
//		}
//	}
//    pool.shutdown();
//}
//	
//	/**
//	 * 并发执行
//	 * @param cmd
//	 * @param ssh 
//	 * @author gulixing@msn.com
//	 * @date   2016年7月19日
//	 */
//	@Deprecated
//	private static void conExec(String cmd, SshUtil ssh) {
//		ListenableFuture<String> future = pool.submit(new Callable<String>() {
//		    @Override
//		    public String call() throws Exception {
//		    	return exec(cmd, ssh);
//		    }
//		});
//
//		Futures.addCallback(future, new FutureCallback<String>() {
//		  @Override
//		  public void onSuccess(String result) {
//			  logger.info("\n------result of ["+ssh.getHost().getAddr()+"]:----------");
//		      System.out.println(result+"\n---------------------");
//		  }
//
//		  @Override
//		  public void onFailure(Throwable t) {
//		      logger.error("主机:"+ssh.getHost().getAddr()+"命令执行失败!",t);
//		  }
//        });
//	}
//
//	@Deprecated
//	private static String exec(String cmd, SshUtil ssh){
//		try {
//			return ssh.exec(cmd).result();
//		} catch (Exception e) {
//		    logger.error("主机:"+ssh.getHost().getAddr()+"命令执行失败!",e);
//		}
//		return "failure";
//	}
}